package zookeeper.testcurator1;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.api.CuratorListener;
import org.apache.curator.framework.api.CuratorWatcher;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.data.Stat;
import org.junit.Before;
import org.junit.Test;

/**
 * Refference https://blog.csdn.net/wo541075754/article/details/69138878
 * 也可以参考官网，官网讲的比较详细
 *
 * Created by majian8 on 2018/6/18.
 */
public class TestClient {
    public CuratorFramework client;
    @Before
    public void setUp(){
        client=ClientUtils.connect();
    }

    /**
     * 创建节点
     */
    @Test
    public void test1() throws Exception {
        //创建一个初始内容为空的节点
        client.create().forPath("/test1");
        //创建一个带内容的节点
        client.create().forPath("/test2","123".getBytes());

    }

    /**
     * 创建临时节点并递归创建父节点
     */
    @Test
    public void test2() throws Exception {
        client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/test3/test33");
    }

    /**
     * 注意 在创建client的时候，设置了一个 namespace,(相当于一个父节点)
     *也就是说下面的test1 路径其实 是/demo/test1
     * 删除节点
     */
    @Test
    public void test3() throws Exception {
        client.delete().forPath("/test1");
    }

    /**
     * 递归的删除节点和其下的子节点
     */
    @Test
    public void test4() throws Exception {
        client.delete().deletingChildrenIfNeeded().forPath("/test3");
    }

    /**
     * 指定版本进行删除
     */
    @Test
    public void test5() throws Exception {
        client.delete().withVersion(1).forPath("/test2");

    }

    /**
     * 强制保证删除
     * 只要客户端会话有效，那么Curator会在后台持续进行删除操作，
     * 直到节点删除成功。比如遇到一些网络异常的情况，此guaranteed的强制删除就会很有效果
     */
    @Test
    public void test6() throws Exception {
        client.delete().guaranteed().forPath("test");
    }

    //##########################读取数据#############################################

    /**
     * 读取节点数据内容API相当简单，Curator提供了传入一个Stat，
     * 使用节点当前的Stat替换到传入的Stat的方法，查询方法执行完成之后，
     * Stat引用已经执行当前最新的节点Stat
     */
    @Test
    public void test7() throws Exception {
        byte [] bt1=client.getData().forPath("/test1");
        System.out.println("获取的数据："+new String(bt1));
        //获取stat
        Stat stat=new Stat();
        byte [] bt2=client.getData().storingStatIn(stat).forPath("/test1");
        System.out.println("获取的stat信息为："+stat.getCversion());

    }

    //##########################更新数据#############################################

    @Test
    public void test8() throws Exception {
        client.setData().forPath("/test1","abc".getBytes());
        //指定版本删除
        //client.setData().withVersion(1).forPath("/test1","def".getBytes());
    }

    /**
     * 异步接口
     *
     * 在使用以上针对节点的操作API时，我们会发现每个接口都有一个inBackground()方法可供调用。
     * 此接口就是Curator提供的异步调用入口。对应的异步处理接口为BackgroundCallback。
     * 此接口指提供了一个processResult的方法，用来处理回调结果。
     * 其中processResult的参数event中的getType()包含了各种事件类型，
     * getResultCode()包含了各种响应码
     * public T inBackground(BackgroundCallback callback, Executor executor);
     * @throws Exception
     */
    @Test
    public void test() throws Exception {


    }

    //###################### 监听器  #################################################

    /**
     * 第一种：Watch 监听请
     * 利用Watcher来对节点进行监听操作，但此监听操作只能监听一次，与原生API并无太大差异。
     * 如有典型业务场景需要使用可考虑，但一般情况不推荐使用
     * @throws Exception
     */
    @Test
    public void test9() throws Exception {
        byte [] bt=client.getData().usingWatcher(new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                System.out.println("节点改动触发监听器-------》"+watchedEvent);
            }
        }).forPath("/test1");

        System.out.println("被监听的节点的内容======》"+new String(bt));
        //第一次变更节点数据
        client.setData().forPath("/test1","aaaa11".getBytes());
        //第二次变更节点内容
        client.setData().forPath("/test2","bbbbb2222".getBytes());
        //删除也会触发监听器
        //client.delete().forPath("/test1");
        Thread.sleep(5000);
        client.close();
    }

    /**
     * 第二种：CuratorWatcher 监听器
     * CuratorListener监听，此监听主要针对background通知和错误通知。使用此监听器之后，
     * 调用inBackground方法会异步获得监听，而对于节点的创建或修改则不会触发监听事件
     *
     * 这里有疑问，监听器触发了3 次？？？？？？？？？？？？？？？？？？？？
     */
    @Test
    public void test10() throws Exception {
        CuratorListener listener=new CuratorListener() {
            @Override
            public void eventReceived(CuratorFramework curatorFramework, CuratorEvent curatorEvent) throws Exception {
                System.out.println("触发监听事件："+curatorEvent);
            }
        };

        //添加监听器
        client.getCuratorListenable().addListener(listener);
        //异步获取节点数据
        System.out.println("获取节点数据 start......");
        client.getData().inBackground().forPath("/test2");
        System.out.println("获取节点数据 end......");
        //变更节点数据
        System.out.println("变更节点数据 start ------》");
        client.setData().forPath("/test2","456".getBytes());
        System.out.println("变更节点数据结束-------》");
        Thread.sleep(10000);
        client.close();
    }

    /**
     * 第三种：
     * Curator引入了Cache来实现对Zookeeper服务端事件监听，
     * Cache事件监听可以理解为一个本地缓存视图与远程Zookeeper视图的对比过程。
     * Cache提供了反复注册的功能。Cache分为两类注册类型：节点监听和子节点监听。
     * NodeCache
     * 用于监听数据节点本身的变化。提供了两个构造方法：
     * public NodeCache(CuratorFramework client, String path)
     * public NodeCache(CuratorFramework client, String path, boolean dataIsCompressed)
     * 其中参数dataIsCompressed表示是否对数据进行压缩，而第一个方法内部实现为调用第二个方法，
     * 且dataIsCompressed默认设为false
     */
    @Test
    public void test11() throws Exception {
        final NodeCache cache=new NodeCache(client,"/test2");
        cache.start();
        cache.getListenable().addListener(new NodeCacheListener() {
            @Override
            public void nodeChanged() throws Exception {
                System.out.println("监听事件触发l");
                System.out.println("重新获取节点内容："+new String(cache.getCurrentData().getData()));
            }
        });
        client.setData().forPath("/test2","123".getBytes());
        client.setData().forPath("/test2","456".getBytes());
        client.setData().forPath("/test2","789".getBytes());
    }
}
